package com.learn.kafka.stream;

import com.learn.kafka.stream.config.ProducerProperties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.concurrent.TimeUnit;

/**
 * 生产者：向输入主题发送文本数据，供Kafka Stream处理
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/8/9 16:46
 * @updateUser:     MI
 * @updateTime:     2025/8/9 16:46
 * @updateRemark:   修改内容
 * @version:        1.0
 */
public class StreamDemoProducer extends ProducerProperties  {

    public static void main(String[] args) throws InterruptedException {
        new StreamDemoProducer().sendMessages();
    }

    // 构造方法：初始化当前生产者的特殊配置
    public StreamDemoProducer() {
        super(); // 调用父类构造方法，加载通用配置
        initProducerConfig();
    }

    // 初始化当前生产者的特殊配置（覆盖或补充父类配置）
    private void initProducerConfig() {
        // 覆盖键序列化器为IntegerSerializer（父类默认是StringSerializer）
//        setConfig("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        setConfig(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 如需修改其他配置（如acks策略），可在此处添加
        // setConfig("acks", "all");
    }

    // 发送消息的核心逻辑
    private void sendMessages() throws InterruptedException {
        // 从父类获取生产者实例（泛型指定为<Integer, String>）
        Producer<String, String> producer = new KafkaProducer<>(super.getProps());
        String topic = "text-input";

        // 发送测试数据
        String[] messages = {
                "Kafka Stream is a powerful stream processing library",
                "Stream processing allows real-time data handling",
                "Kafka is used for building real-time data pipelines",
                "Stream processing can transform and aggregate data",
                "Kafka Stream API is easy to use with Java"
        };

        for (String msg : messages) {
            producer.send(new ProducerRecord<>(topic, null, msg));
            System.out.println("发送消息: " + msg);
            TimeUnit.SECONDS.sleep(1); // 间隔1秒发送
        }

        // 关闭生产者
        producer.close();
    }
}
